package m4;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

public class Producer {
    public static void main(String[] args) throws MQClientException {
        //事务消息生产者
        TransactionMQProducer p = new TransactionMQProducer("producer-group3");
        //连接name server
        p.setNamesrvAddr("192.168.64.141:9876");

        //设置事务消息监听器，发送事务消息后会自动触发监听器执行本地事务
        p.setTransactionListener(new TransactionListener() {
            Map<String,LocalTransactionState> map = new ConcurrentHashMap<>();

            //执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("执行本地事务");
                if(Math.random()<0.33){
                    System.out.println("本地事务执行成功");
                    map.put(message.getTransactionId(),LocalTransactionState.COMMIT_MESSAGE);
                    return LocalTransactionState.COMMIT_MESSAGE;//提交消息
                }else if(Math.random()<0.33){
                    System.out.println("本地事务执行失败");
                    map.put(message.getTransactionId(),LocalTransactionState.ROLLBACK_MESSAGE);
                    return LocalTransactionState.ROLLBACK_MESSAGE;//回滚消息
                }else {
                    System.out.println("本地事务执行情况未知");
                    map.put(message.getTransactionId(),LocalTransactionState.UNKNOW);
                    return LocalTransactionState.UNKNOW;
                }
            }

            //处理rocketmq的事务回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("服务器正在回查消息状态");
                return map.get(messageExt.getTransactionId());

            }
        });

        //启动
        p.start();
        //发送消息
        while (true){
            System.out.println("输入消息：");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());
            //发送事务消息，会触发本地事务执行
            //p.sendMessageInTransaction(msg,执行本地事务时使用的业务数据对象);
            System.out.println("发送半消息");
            p.sendMessageInTransaction(msg,null);
            System.out.println("事务消息已发送");
        }

    }
}
